package com.cheng.im.gateway.zk;

import com.alibaba.fastjson2.JSONObject;
import com.cheng.im.common.bean.msg.ProtoMsg;
import com.cheng.im.common.constants.ServerConstants;
import com.cheng.im.common.entity.ImNode;
import com.cheng.im.common.utils.ObjectUtil;
import com.cheng.im.common.utils.ThreadUtil;
import com.cheng.im.gateway.client.GateWayRouterClient;
import com.cheng.im.gateway.entity.ImWorker;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

/**
 * 监听其他节点的上下线
 * @author Cheng
 */
@Slf4j
@Component
public class GateWayRouterListen {

    @Resource(name = "curatorZKClient")
    private CuratorFramework client;

    private String pathRegistered = null;

    private ImNode node = null;

    private static final String path = ServerConstants.MANAGE_PATH;

    private ConcurrentHashMap<Long, GateWayRouterClient> workerMap =
            new ConcurrentHashMap<>();


    /**
     * 节点上线执行
     */
    private BiConsumer<ImNode, GateWayRouterClient> runAfterAdd = (node, relaySender) -> {
        doAfterAdd(node, relaySender);
    };

    /**
     * 节点下线执行
     */
    private Consumer<ImNode> runAfterRemove = (node) -> {
        doAfterRemove(node);
    };

    @PostConstruct
    public void init() {
        try {
            // 订阅节点的增加和删除事件
            PathChildrenCache childrenCache = new PathChildrenCache(client, path, true);
            PathChildrenCacheListener childrenCacheListener = new PathChildrenCacheListener() {
                @Override
                public void childEvent(CuratorFramework client,
                                       PathChildrenCacheEvent event) throws Exception {
                    log.info("===============监听到节点变化=================");
                    ChildData data = event.getData();
                    switch (event.getType()) {
                        case CHILD_ADDED:
                            log.info("zk 子节点增加 : " + data.getPath() + "  数据:" + data.getData());
                            processNodeAdded(data);
                            break;
                        case CHILD_REMOVED:
                            log.info("zk 子节点下线 : " + data.getPath() + "  数据:" + data.getData());
                            processNodeRemoved(data);
                            break;
                        case CHILD_UPDATED:
                            log.info("zk 子节点更新 : " + data.getPath() + "  数据:" + new String(data.getData()));
                            break;
                        default:
                            log.debug("zk [PathChildrenCache]节点数据为空, path={}", data == null ? "null" : data.getPath());
                            break;
                    }

                }
            };

            childrenCache.getListenable().addListener(
                    childrenCacheListener, ThreadUtil.getIoIntenseTargetThreadPool());
            childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);

            log.info("===================zk节点监听已经启动====================");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void processNodeRemoved(ChildData data) {

        byte[] payload = data.getData();
        ImNode node = ObjectUtil.JsonBytes2Object(payload, ImNode.class);


        long id = ImWorker.getInst().getIdByPath(data.getPath());
        node.setId(id);
        log.info("zk [TreeCache]节点删除, path={}, data={}",
                data.getPath(), JSONObject.toJSONString(node));

        if (runAfterRemove != null) {
            runAfterRemove.accept(node);
        }


    }

    private void doAfterRemove(ImNode node) {
        GateWayRouterClient gateWayRouterClient = workerMap.get(node.getId());

        if (null != gateWayRouterClient) {
            gateWayRouterClient.stopConnecting();
            workerMap.remove(node.getId());
        }
    }

    /**
     * 节点增加的处理
     *
     * @param data 新节点
     */
    private void processNodeAdded(ChildData data) {
        byte[] payload = data.getData();
        ImNode node = ObjectUtil.JsonBytes2Object(payload, ImNode.class);

        long id = ImWorker.getInst().getIdByPath(data.getPath());
        node.setId(id);

        log.info("zk [TreeCache]节点更新端口, path={}, data={}",
                data.getPath(), JSONObject.toJSONString(node));

        if (node.equals(getLocalNode())) {
            log.info("zk [TreeCache]本地节点, path={}, data={}",
                    data.getPath(), JSONObject.toJSONString(node));
            return;
        }
        GateWayRouterClient relaySender = workerMap.get(node.getId());
        // 重复收到注册的事件
        if (null != relaySender && relaySender.getRmNode().equals(node)) {

            log.info("zk [TreeCache]节点重复增加, path={}, data={}",
                    data.getPath(), JSONObject.toJSONString(node));
            return;
        }

        if (runAfterAdd != null) {
            runAfterAdd.accept(node, relaySender);
        }
    }


    private void doAfterAdd(ImNode n, GateWayRouterClient relaySender) {
        if (null != relaySender) {
            //关闭老的连接
            relaySender.stopConnecting();
        }
        //创建一个消息转发器
        relaySender = new GateWayRouterClient(n);
        //建立转发的连接    todo: 还是建立了连接
        relaySender.doConnect();

        workerMap.put(n.getId(), relaySender);
    }


    public GateWayRouterClient route(long nodeId) {
        GateWayRouterClient gateWayRouterClient = workerMap.get(nodeId);
        if (null != gateWayRouterClient) {
            return gateWayRouterClient;
        }
        return null;
    }


//    public void sendNotification(String json) {
//        workerMap.keySet().stream().forEach(
//                key ->
//                {
//                    if (!key.equals(getLocalNode().getId())) {
//                        GateWayRouterClient peerSender = workerMap.get(key);
//                        ProtoMsg.Message pkg = NotificationMsgBuilder.buildNotification(json);
//                        peerSender.writeAndFlush(pkg);
//                    }
//                }
//        );
//
//    }

    public ImNode getLocalNode() {
        return ImWorker.getInst().getLocalNodeInfo();
    }

    public void remove(ImNode remoteNode) {
        workerMap.remove(remoteNode.getId());
        log.info("[TreeCache]移除远程节点信息,  node={}", JSONObject.toJSONString(remoteNode));
    }

}
